package com.example.bigdata.spark.RDD练习

import org.apache.spark.{SparkConf, SparkContext}

object RDDTest {

    /**
      * 练习RDD 算子，加深理解算子的功能
      * 问题如下：
      * 1、一共有多少人参加考试？
      * 2、一共有多少个小于20岁的人参加考试？
      * 3、一共有多少个等于20岁的人参加考试？
      * 4、一共有多少个大于20岁的人参加考试？
      * 5、一共有多个男生参加考试？
      * 6、一共有多少个女生参加考试？
      * 7、12班有多少人参加考试？
      * 8、13班有多少人参加考试？
      * 9、语文科目的平均成绩是多少？
      * 10、数学科目的平均成绩是多少？
      * 11、英语科目的平均成绩是多少？
      * 12、单个人平均成绩是多少？
      * 13、12班平均成绩是多少？
      * 14、12班男生平均总成绩是多少？
      * 15、12班女生平均总成绩是多少？
      * 16、同理求13班相关成绩?
      * 17、全校语文成绩最高分是多少？
      * 19、12班语文成绩最低分是多少？
      * 20、13班数学最高成绩是多少？
      * 21、总成绩大于150分的12班的女生有几个？
      *     13 王英 19 女 english 70
      */
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
                .setAppName("RDD 练习")
                .setMaster("local[1]")
        val sc = new SparkContext(conf)
        val file = "input/RDDtext.txt"
        val atmFile = "input/atm_map"
        //atm(sc,atmFile)
        /*nb1(sc,file)
        nb2(sc,file)
        nb3(sc,file)
        nb4(sc,file)
        nb5(sc,file)
        nb6(sc,file)
        nb7(sc,file)
        nb8(sc,file)
        nb9(sc,file)*/
        nb10(sc,file)
        sc.stop()
    }
    // 求 班级的平均分
    def nb10(sc: SparkContext, file: String): Unit ={
        val rdd1 = sc.textFile(file)
                .map(x=>{
                    val fields = x.split(" ")
                    // (classid,name,age,sex,course,score)
                    (fields(0),fields(1),fields(2),fields(3),fields(4),fields(5))
                })
        //每个人的平均分
        val stu_avg = rdd1.map(x => (x._2,x._6.toDouble)).groupByKey(1).map(x=>(x._1,x._2.sum/x._2.size))
        stu_avg.foreach(println)
        // 加入班级字段
        val stu_class = rdd1.map(x=> (x._2,x._1)).distinct()
        stu_class.foreach(println)
        val stu = stu_class.join(stu_avg).map(x=>(x._2._1,x._1,x._2._2))
        stu.foreach(println)
    }
    //一共有多少人参加考试？
    def nb1(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file)
        // 格式化 每行
        val rdd2 = rdd1.map(line => {
            val fields = line.split(" ")
            val classid = fields(0)
            val name = fields(1)
            val age = fields(2)
            val sex = fields(3)
            val course = fields(4)
            val score = fields(5)
            (classid,name,age,sex,course,score)
        })
        rdd2.cache()
        // 过滤关键列，去重，统计数量
        val rdd3 = rdd2.map(_._2).distinct().count()
        println(rdd3.toString)

    }
    //一共有多少个小于20岁的人参加考试？
    def nb2(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file)
        // 格式化 每行
        val rdd2 = rdd1.map(line => {
            val fields = line.split(" ")
            val classid = fields(0)
            val name = fields(1)
            val age = fields(2)
            val sex = fields(3)
            val course = fields(4)
            val score = fields(5)
            (classid,name,age,sex,course,score)
        })
        val rdd3 = rdd2.filter(x => {
         x._3.toInt > 20
        }).map(x=>(x._2,x._3)).distinct()
        rdd3.foreach(println(_))
    }
    //一共有多个男生参加考试？
    def nb3(sc:SparkContext,file:String): Unit ={
        val rdd1 = sc.textFile(file).map(line=>{
            val fields = line.split(" ")
            val classid = fields(0).toInt
            val name = fields(1)
            val age = fields(2).toInt
            val sex = fields(3)
            val course = fields(4)
            val score = fields(5).toInt
            (name,sex)
        }).distinct().filter(_._2 == "男")
        rdd1.foreach(println(_))
    }
    //每班有多少人参加考试？
    def nb4(sc:SparkContext,file: String): Unit =
    {
        val rdd1 = sc.textFile(file).map(line => {
            val fields = line.split(" ")
            (fields(0),fields(1))
        }).distinct().groupByKey().mapValues(_.size)
        rdd1.foreach(println(_))
    }
    // 语文科目的平均成绩是多少？
    def nb5(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file).map(line => {
            val fields = line.split(" ")
            (fields(4),fields(5).toDouble)
        }).filter(_._1 == "chinese")

        val rdd2 = rdd1.map(_._2).mean()
        println("rdd2 : "+rdd2)
        // 方法2
        /*val rdd12 = rdd1.groupByKey().mapValues(_.toList).mapValues(x=>x.sum/x.size)
        rdd12.foreach(println(_))*/

    }
    //单个人平均成绩是多少？
    def nb6(sc:SparkContext,file: String): Unit =
    {
        val rdd1 = sc.textFile(file).map(line => {
            val fields = line.split(" ")
            (fields(1),fields(5).toDouble)
        })
        val rdd2 = rdd1.groupByKey()
        /*val rdd3 = rdd2.mapValues(_.toList)
        val rdd4 = rdd3.mapValues(x => x.sum/x.size)
        rdd4.foreach(println(_))*/
        val rdd12 = rdd2.map(x => (
            x._1,(x._2.sum/x._2.size)
        ))
        rdd12.foreach(println(_))
    }
    // 12班平均成绩是多少？
    def nb7(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file).map(line =>{
            val feilds = line.split(" ")
            (feilds(0),feilds(5).toDouble)
        })
        val rdd2 = rdd1.filter(_._1 == "12").groupByKey()
                        .map(x=>{
                            (x._1,x._2.sum,x._2.size)
                        })
        rdd2.foreach(println(_))
        val rdd3 = rdd1.filter(_._1 == "12").map(_._2).mean()
        println("rdd3 : " + rdd3)
        val rdd4 = rdd1.filter(_._1.equals("12")).map(_._2.toInt).mean()
        println("rdd4 : " + rdd4)
    }
    //全校语文成绩最高分是多少？70
    def nb8(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file).map(x => {
            val fields = x.split(" ")
            (fields(4),fields(5))
        })
        val rdd2 = rdd1.filter(_._1 == "chinese").map(_._2).max()
        println("rdd2 : " + rdd2)
    }
    // 总成绩大于150分的12班的女生有几个？
    def nb9(sc:SparkContext,file:String): Unit =
    {
        val rdd1 = sc.textFile(file).map(line =>{
            val fields = line.split(" ")
            (fields(0),fields(1),fields(3),fields(5).toInt)
        })
        val rdd2 = rdd1.filter( x=> (x._1 =="12" && x._3 == "女"))
                .map(x => (x._1,x._4)).groupByKey()
                .map(x =>(x._1,x._2.sum))
                .filter(_._2 > 150)
        rdd2.foreach(println(_))
    }
    def atm(sc:SparkContext,atmFile:String): Unit =
    {
        val rdd1 = sc.textFile(atmFile).map(_.split("\\001"))
        // 查看 行内容
        rdd1.foreach(x=>(println(x.mkString("->"))))
    }
}
